package net.wlm.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.wlm.jtp.dwd.log.util.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用state编程，对is_new进行校验
 */

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String, String, String> {
    // a-声明状态
    private transient ValueState<String> firstDateState;
    @Override
    public void open(Configuration parameters) throws Exception {
        // b-初始化状态
        firstDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("first-date", String.class)
        );
    }
    @Override
    public void processElement(String s, Context ctx, Collector<String> out) throws Exception {
        // c-获取状态值
        String firstDate = firstDateState.value();
        // 解析字段值
        JSONObject jsonObject = JSON.parseObject(s);
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        Long tsValue = jsonObject.getLong("ts");
        // isNew = 1 --> firstDate = null , firstDate = 有值
        if("1".equals(isNewValue)){
            if(null == firstDate){
                // d-更新状态值
                // 此时 is_new正确，状态值更新为第一次访问日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            }else{
                // 状态值（首次访问日期）等于数据日期，不做处理
                if(!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                    // 此时  is_new错误，更新状态值
                    jsonObject.getJSONObject("common").put("is_new","0");
                }
            }
        }
        else {
            // 当用户早早登录访问，当实时应用后来处理，用户第一次访问时，firstDate为null
            if(null == firstDate){
                // 设置昨天日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue - 86400000, DateTimeUtil.DATE_FORMAT));
            }
        }
        // 输出数据
        out.collect(jsonObject.toJSONString());
    }
}